package main

import (
	"context"
	"database/sql"
	"encoding/json"
	"fmt"
	"log"

	_ "github.com/go-sql-driver/mysql"
	"github.com/segmentio/kafka-go"
)

func main() {
	// 初始化 MySQL 连接
	dsn := fmt.Sprintf("%s:%s@tcp(%s:%d)/%s", config.Mysql.User, config.Mysql.Pass, config.Mysql.Host, config.Mysql.Port, config.Mysql.Database)
	db, err := sql.Open("mysql", dsn)
	if err != nil {
		log.Fatalf("Failed to connect to MySQL: %v", err)
	}
	defer db.Close()

	// 测试 MySQL 连接
	err = db.Ping()
	if err != nil {
		log.Fatalf("Failed to ping MySQL: %v", err)
	}

	// 初始化 Kafka 读取器
	r := kafka.NewReader(kafka.ReaderConfig{
		Brokers: []string{config.Kafka.Server},
		Topic:   config.Kafka.Topic,
		GroupID: "user_follow_consumer_group",
	})
	defer r.Close()

	mysqlTable := "user_follow"
	fmt.Println("Starting to consume messages from Kafka...")
	for {
		msg, err := r.FetchMessage(context.Background())
		if err != nil {
			log.Fatalf("Failed to fetch message from Kafka: %v", err)
		}

		var followMsg FollowMessage
		err = json.Unmarshal(msg.Value, &followMsg)
		if err != nil {
			log.Printf("Failed to unmarshal message: %v", err)
			continue
		}

		if followMsg.Action == "user_follow" {
			// 将消息写入 MySQL
			_, err = db.Exec(fmt.Sprintf("INSERT INTO %s (follower_id, followed_id) VALUES (?, ?)", mysqlTable), followMsg.FollowerID, followMsg.FollowedID)
			if err != nil {
				log.Printf("Failed to insert record into MySQL: %v", err)
			} else {
				fmt.Printf("Inserted user follow record into MySQL: %d followed %d\n", followMsg.FollowerID, followMsg.FollowedID)
			}
		}
	}
}
